#include <sys/socket.h>
#include <netinet/in.h>
#include <errno.h>
#include <sys/types.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <fcntl.h>
#include "VAC_video.h"
#include "link.h"
#include "common.h"
#include "msg_core.h"
#include "user_core.h"
#include "queue.h"

#define VACVIDEO_BASEPATH        "/upload/VACVideo"
#define VIDEO_MSG_SERVER_IP      "192.168.61.2"
#define VIDEO_MSG_SERVER_PORT    7000
#define VIDEO_DATA_PORT  	  8005

#define _15Min		(15 * 60) //15min
#define _24Hour	(24*60*60)


#define VIDEO_MSG_SOCKET_SYS_CACHE_SIZE		(2*1024*1024)
#define VIDEO_DATA_SOCKET_SYS_CACHE_SIZE	(6*1024*1024)
#define VIDEO_MSG_IDEX_BUFFER_MAX_SIZE		128
#define VIDEO_MSG_IDEX_BUFFER_COUNT             64
#define VIDEO_MSG_MAX_SIZE     		4096    //4KB
#define VIDEO_MSG_MIN_SIZE		14
#define _VIDEO_MSG_HEADER1_    		0xAA
#define _VIDEO_MSG_HEADER2_    		0xAA
#define VIDEO_DATA_MAX_CLIENT	        VIDEO_MSG_MIN_SIZE

#define VIDEO_REGISTER_timeout		10
#define VIDEO_HEARTBEAT_interval	30
#define VIDEO_HEARTBEAT_timeout		60
#define VIDEODEV_timeout		10

#define CMD_INIT		0x00
#define CMD_WAITTING		0x01
#define CMD_OK			0x02
#define CMD_NOTIFY		0x01
#define CMD_STARTING		0x02

#define VIDEO_CHANNEL_MAX_COUNT	   14
//#define VIDEO_SW_VERSION	"V1.00   "
#define VIDEO_DEVICE_ID		"0000001"
#define VIDEO_VERSION_LENGHT	8
#define VIDEOMSG_OK		1
#define VIDEOMSG_FAIL		0

#define VIDEO_DEVICE_SELF_MFRS_ID   0x02

#define LAST_NULL   	        0xF0
#define VIDEO_DATA_PACKET_BUFFER_COUNT	      128
#define VIDEO_DATA_PACKET_BUFFER_SIZE	      1500
#define VIDEO_DATA_PACKET_MAX_SIZE	      1419
#define VIDEO_DATA_PACKET_QUEUE_MAX_COUNT     3
#define VIDEO_DATA_PACKET_HEADER_START_POS    8
#define VIDEO_DATA_TYPE_POS                   6
#define VIDEO_DATA_CHANNEL_POS               16
#define VIDEODATA_START_POS                  19

#define VIDEO_DATA_MAX_SIZE           (144*1024)    //144kB
#define VIDEO_DATA_MAX_COUNT          32

#define VIDEO_SUB_PTHREAD_STACK_SIZE (3*1024*1024) //2MB
typedef struct __videoPacket_head{
    UINT32 id;
    UINT8  isPacket;
    UINT8  total;
    UINT8  index;
}VIDEOPACKET_HEAD;

typedef struct __videoPacket_data{
    UINT8  valid;
    UINT16 len;
    CHAR data[VIDEO_DATA_PACKET_BUFFER_SIZE];
}VIDEOPACKET_DATA;
typedef struct __videoDataPthread_share{
    QUEUE_DESC dataList;
   // pthread_mutex_t endLock; 
    CHAR endFlg;
}VIDEO_DATA_PTHREAD_SHARE;
typedef struct __videoPacket_info{
    VIDEOPACKET_DATA data[VIDEO_DATA_PACKET_BUFFER_COUNT];
    UINT32 id;    
    UINT8  total;
    UINT32 size;
    UINT8  valid;
}VIDEOPACKET_INFO;

typedef struct __manufacturer_Info{//changshang info
    CHAR ID;      
    CHAR name[20];
    CHAR version[8];      
} MANUFACTURER_INFO; 
typedef struct __train_info{//jiche info
    CHAR ID[16];
    CHAR number[16]; 
} TRAIN_INFO;
typedef struct {
    CHAR ID;         //channel id
    CHAR valid;    
    CHAR name[64];   //channel name
}VIDEO_CHANNEL_INFO;
typedef struct {
    CHAR filename[256];  //record filename
    CHAR backFilename[256];//dst filename
    INT32 fd;         //record file fd
} FILE_DESC;
typedef struct __video_channel_into{
    VIDEO_CHANNEL_INFO ch;
    CHAR status;     //if start record data
    FILE_DESC file;  //cur file
    INT32 socket;     //client socket    
    UINT32 curTimeInt;
} VIDEO_CHANNEL_CTRL;
typedef struct{
    INT32 socket;
    UINT32 curPacketId;
    UINT8 channel;
    pthread_t pid;    
} VIDEODataChannel;
typedef struct __video_ctrl{
    VIDEO_CHANNEL_CTRL CHsctrl[VIDEO_CHANNEL_MAX_COUNT];  //channels info
    FILE_DESC idexFile;
    CHAR strTime[16]; //filename date time string
    CHAR status;      //if get channel list
    CHAR validSize;   //size of valid channel 
    CHAR lock;
} VIDEO_CTRL_INFO;
typedef struct _SOCKET_INFO_{
    CHAR ip[16];
    UINT16 port;    
}SOCKET_INFO;

typedef struct __device_info{
    TRAIN_INFO train;
    MANUFACTURER_INFO  mfrs; 
    VIDEO_CTRL_INFO videoCtrl;
    SOCKET_INFO msgTcp;
    SOCKET_INFO dataTcp;
    UINT16 dataBackPort;
    CHAR backPath[256];
    CHAR path[256];
    UINT32 curDayInt;    
    INT32 socket;
    INT32 sSocket;
    CHAR regStatus;  
    CHAR hStatus;
    //CHAR reqStatus;
    CHAR status;    
    CHAR ssdStatus;
    CHAR uploadStatus;
    CHAR idStatus;
      
}device_info;

typedef struct __video_msg_header{
    CHAR head1;
    CHAR head2;
    UINT16 size;    
}video_msg_header;

typedef enum {
    MSG_NULL = 0,			//没有消息类型
    MSG_REG = 0x01,			//注册 -->
    MSG_REG_ACK,	
    MSG_MFRS_ID = 0x03,	                //<--
    MSG_MFRS_ID_ACK,		
    MSG_STATUS = 0x05,		        //-->
    MSG_STATUS_ACK,	
    MSG_TIME_SYNC = 0x07,	        //XX
    MSG_TIME_SYNC_ACK,		        //XX
    MSG_GET_VIDEO_CHANNEL = 0x09,       //-->
    MSG_GET_VIDEO_CHANNEL_ACK,
    MSG_REQ_VIDEO_CHANNEL = 0x0B,       //-->
    MSG_REQ_VIDEO_CHANNEL_ACK,
    MSG_REQ_OFF_VIDEO_CHANNEL = 0x0D,   //XX
    MSG_REQ_OFF_VIDEO_CHANNEL_ACK,      //XX
    MSG_HEART = 0x0F,                   //XX
    MSG_HEART_ACK,			//XX
    MSG_COM_SAFE = 0x11,		//<--
    MSG_VIDEO_DATA_CHANNEL_NEW = 0x12,		//-->
    MSG_VIDEO_DATA_CHANNEL_NEW_ACK,	
    MSG_VIDEO_DATA = 0x14,		//<--
    MSG_SIZE			//未定义消息
} VIDEO_MSG_TYPE_E;

device_info videoDevInfo;
QUEUE_DESC videoMsgDesc;
extern AIR_APP_INFO_T stMqttMsgInfo;
CHAR *socket_type[] = {"MSG","DATA"};
INT32 videoMSG_recv(int socket ,CHAR *data, int size );
INT32 videoData_recv(int socket ,CHAR *data, int size );
INT32 videoMSG_deCode(const CHAR*data , INT32 size ,CHAR *dst , INT32 *type );
INT32 videoMSG_send(INT32 socket , CHAR *data, INT32 size);
INT32 idexProcess_task (void *arg);

MANUFACTURER_INFO MFRS_lists[] = {
    {1         , "株洲所"      ,""} ,
    {2         , "成都运达"     ,""},
    {3         , "北京蓝天多维" ,""},
    {4         , "济南若临"     ,""},
    {5 	       , "郑州思富特"   ,""},
    {LAST_NULL    , "其他厂商"          ,""}
};
VIDEO_CHANNEL_INFO CHALLEL_lists[] = {
    {1      ,0   , "一端路况"} ,//"1unknown"},//
    {2      ,0   , "一端司机室" },
    {3      ,0   , "机械间1"},
    {4      ,0   , "机械间2"},
    {5 	    ,0   , "机械间3"},
    {6 	    ,0   , "机械间4" },
    {7 	    ,0   , "unknown" },
    {8 	    ,0   , "unknown"},
    {9 	    ,0   , "二端路况" },
    {10     ,0   , "二端司机室" },
    {11	    ,0   , "unknown" },
    {12     ,0   , "unknown"  },
    {13     ,0   , "unknown" },
    {14     ,0   , "unknown"  },
    {LAST_NULL   ,0  , ""   }
};
void void_print( CHAR *data, int size ,int flg){
    //return;
    int i = 0;
    if(flg == 0)
        printf("send[%d]:",size);
    else
        printf("recv[%d,%d]:",size,flg);
    if(size > 128)
        size = 128;
    for(i = 0 ;i < size ; i++)
    {
        printf("%02x ",(UINT8)data[i]);
    }
    printf("\n");

}
INT32 video15Min_check(INT32 channel)
{
    UINT32 ct = currentTimeGet_int() / _15Min;
    
    if(channel <= 0)
        return ERROR;
    if(videoDevInfo.videoCtrl.CHsctrl[channel - 1].curTimeInt == ct)
        return ERROR;
    return OK;
}
INT32 videoNextDay_check(void)
{
   UINT32 cday = currentTimeGet_int() / _24Hour;
   if(videoDevInfo.curDayInt == cday)
       return ERROR;
    return OK;
}
void videoPath_check(void)
{
    CHAR dayBuf[256] = {0};
    struct tm p_tm = {0};
    time_t timep;

    time(&timep);
    localtime_r(&timep,&p_tm); 

    if(videoDevInfo.regStatus != CMD_OK || videoDevInfo.idStatus != CMD_OK) 
        return;
    if(access(videoDevInfo.path,F_OK) != 0){
        INFO("<video> mkdir -p %s",videoDevInfo.path);
        htkj_system("mkdir -p %s",videoDevInfo.path);
    }
    if(access(videoDevInfo.backPath,F_OK) != 0){       
        htkj_system("mkdir -p %s",videoDevInfo.backPath);
    }
    sprintf (dayBuf, "%s/%04d-%02d-%02d",videoDevInfo.path,p_tm.tm_year + 1900, p_tm.tm_mon+ 1,p_tm.tm_mday);

    if(access(dayBuf,F_OK) != 0){
        INFO("<video> mkdir -p %s",dayBuf);
        htkj_system("mkdir -p %s",dayBuf);
    }
    sprintf (dayBuf, "%s/%04d-%02d-%02d",videoDevInfo.backPath,p_tm.tm_year + 1900, p_tm.tm_mon+ 1,p_tm.tm_mday);

    if(access(dayBuf,F_OK) != 0){
        htkj_system("mkdir -p %s",dayBuf);
    }
}
void videoPath_nextDay_check(INT8 *filename)
{
    CHAR dayBuf[256] = {0};
    struct tm p_tm = {0};
    time_t timep;
    CHAR day[20] = {0};

    time(&timep);
    localtime_r(&timep,&p_tm); 

    if(filename == NULL)
        return;
    sprintf(day,"%04d-%02d-%02d",p_tm.tm_year + 1900, p_tm.tm_mon+ 1,p_tm.tm_mday);
    if(strstr(filename,day) != NULL)
        return;
    timep -= 23*3600;
    localtime_r(&timep,&p_tm); 
    if(access(videoDevInfo.backPath,F_OK) != 0){       
        htkj_system("mkdir -p %s",videoDevInfo.backPath);
    }    
    sprintf (dayBuf, "%s/%04d-%02d-%02d",videoDevInfo.backPath,p_tm.tm_year + 1900, p_tm.tm_mon+ 1,p_tm.tm_mday);

    if(access(dayBuf,F_OK) != 0){
        htkj_system("mkdir -p %s",dayBuf);
    }
}
INT32 TCPServer_init(UINT16 port)
{                                                                                                         
    int sock = socket(AF_INET,SOCK_STREAM,0);
    struct sockaddr_in server_socket;    
    const int on = 1;   

    if(sock < 0){   
        Error("<video> Tcp Server init socket");
        return ERROR;
    }
    
    bzero(&server_socket,sizeof(server_socket));
    server_socket.sin_family=AF_INET;
    server_socket.sin_addr.s_addr=htonl(INADDR_ANY);
    server_socket.sin_port=htons(port);

    setsockopt(sock,SOL_SOCKET,SO_REUSEADDR,&on,sizeof(on)); 
    if(bind(sock,(struct sockaddr*)&server_socket,sizeof(struct sockaddr_in)) < 0){
        Error("<video> Tcp Server bind socket");
        close(sock);
        return ERROR;
    }   
    if(listen(sock,VIDEO_DATA_MAX_CLIENT) < 0) {
        Error("<video> Tcp Server listen port[%d]",port);       
        close(sock);
        return ERROR;
    }
    INFO("<video> create Tcp Server success, listen port[%d]",port);  
    return sock;
}
void videoIdxFile_start(FILE_DESC * file)
{
    struct tm p_tm = {0};
    time_t timep;

    time(&timep);
    if(file == NULL){
        Error("<video> idex start file == null");
        return;
    }
    localtime_r(&timep,&p_tm); 
    
    videoPath_check();

    sprintf(file->filename,"%s/%04d-%02d-%02d/%s_%s_%04d%02d%02d.idx.tmp",
        videoDevInfo.path,
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday,
        videoDevInfo.train.ID,
        videoDevInfo.mfrs.name,       
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday                    
    );
    sprintf(file->backFilename,"%s/%04d-%02d-%02d/%s_%s_%04d%02d%02d.idx",
        videoDevInfo.backPath,
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday,
        videoDevInfo.train.ID,
        videoDevInfo.mfrs.name,       
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday                    
    );
    
    file->fd = open(file->filename , O_RDWR  | O_CREAT | O_APPEND,0644);
    if(file->fd <= STDERR_FILENO){
        Error("<video> create file %s,%d fail",file->filename,file->fd);
    }else{
        INFO("<video> create file %s,%d ok",file->filename,file->fd);
    }  
    videoDevInfo.curDayInt = currentTimeGet_int() / _24Hour; 
}
void videoFile_start(FILE_DESC * file , INT32 channel, CHAR *channelName)
{
    struct tm p_tm = {0};
    time_t timep;
    UINT32 ct = currentTimeGet_int();
    UINT32 ct15 = ct / _15Min * _15Min;   

    time(&timep);
    if(file == NULL){
        Error("<video> start file == null");
        return;
    }  
    if(channel <= 0){
        Error("<video> channel[%d] <= 0",channel);
        return;
    }
    if(ct == ct15 || ct == (ct15 + 1))//compare time near 15min
        timep = (time_t)ct15; //
    localtime_r(&timep,&p_tm); 
    
    videoPath_check();
    sprintf(file->filename,"%s/%04d-%02d-%02d/%s_%s_%02d_%s_%04d%02d%02d_%02d%02d%02d.mp4.tmp",//tmp
        videoDevInfo.path,
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday,
        videoDevInfo.train.ID,
        videoDevInfo.mfrs.name,
        channel,
        channelName,
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday,
        p_tm.tm_hour,
        p_tm.tm_min,
        p_tm.tm_sec                
    );
    sprintf(file->backFilename,"%s/%04d-%02d-%02d/%s_%s_%02d_%s_%04d%02d%02d_%02d%02d%02d.mp4",
        videoDevInfo.backPath,
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday,
        videoDevInfo.train.ID,
        videoDevInfo.mfrs.name,
        channel,
        channelName,
        p_tm.tm_year + 1900,
        p_tm.tm_mon+ 1,
        p_tm.tm_mday,
        p_tm.tm_hour,
        p_tm.tm_min,
        p_tm.tm_sec                
    );

    file->fd = open(file->filename , O_RDWR  | O_CREAT | O_APPEND,0644);
    if(file->fd <= STDERR_FILENO){
        Error("<video> create file %s,%d fail",file->filename,file->fd);
    }else{
        INFO("<video> create file %s,%d ok",file->filename,file->fd);
    }     
    videoDevInfo.videoCtrl.CHsctrl[channel - 1].curTimeInt = currentTimeGet_int() / _15Min;
}
void videoFile_end(FILE_DESC * file)
{
    CHAR filename[LINE_CONTENT_MAX_LEN] = {0};
    UINT32 size = 0;
    INT32 ret = -1;

    if(file == NULL){
        Error("<video> end file == null");
        return;
    }
    strcpy(filename,file->filename);
    filename[strlen(filename) - 4] = 0;//del ".tmp"
    if(file->fd > 0){
        close(file->fd);
        file->fd = -1;
        if(access(file->filename,F_OK) != 0)
            return;
        ret = rename(file->filename, filename);     
        videoPath_check();
        videoPath_nextDay_check(file->backFilename);         
        if(ret == 0)
            htkj_system("mv '%s' '%s'",filename, file->backFilename);
        else
            htkj_system("mv '%s' '%s'",file->filename, file->backFilename);
        size = fileSize(file->backFilename);       
    
        INFO ("<video> close file %s size[%d]",file->backFilename ,size);  
    }
    
}
/*void videoIdxFileCP_end(FILE_DESC * file)
{
    if(file->fd > 0){
        close(file->fd);
        file->fd = -1;
        htkj_system("cp -fr %s %s",file->filename, file->backFilename);
    }
}*/
CHAR videoMSG_checksum(const CHAR *data,INT32 size)
{
   INT32 i = 0;
   UINT8 c = 0;
   
   for(i = 0; i < size;i++){
       c += data[i];
   }
   c = (UINT8)_VIDEO_MSG_HEADER1_ - c;
   return c;
}
INT32 fill_version(CHAR * data)
{
    INT32 i = 0;
    INT32 len = 0;
    INT32 pos = 0;

    len = strlen(stMqttMsgInfo.appVer);
    memcpy(&data[pos] , stMqttMsgInfo.appVer, len);
    pos += len;
    if(len < VIDEO_VERSION_LENGHT){
        len = VIDEO_VERSION_LENGHT - len;
        for(i = 0 ; i < len ; i++)
        {
            data[pos++] = ' ';
        }
    }  
    return pos; 
}
INT32 videoMSG_enCode(INT32 type , INT32 ack,CHAR *data ,INT32 size,CHAR *dst, INT32 dSize)
{
    UINT16 pos = 0;
    video_msg_header header;   

    if(dst == NULL || dSize < VIDEO_MSG_MIN_SIZE)
        return 0;
    
    memset(dst, 0 , dSize);
    header.head1 = _VIDEO_MSG_HEADER1_;
    header.head2 = _VIDEO_MSG_HEADER2_;
    
    pos += 4;//skip len
    dst[pos++] = type;
    pos++;
    pos += fill_version(&dst[pos]);   
    if(data != NULL && size > 0){
        if((pos+size+1) < dSize){
            memcpy(&dst[pos] , data,size);
            pos += size;
        }else
            return 0;
    }
    
    if(ack == VIDEOMSG_OK || ack == VIDEOMSG_OK){
        if((pos+2) > dSize){
            return 0;
        }
        dst[pos++] = ack;
    }
    
    //header.size = htons(pos + 1);
    header.size = pos + 1;
    memcpy(&dst[0] ,&header,sizeof(header)); 
    dst[pos] = videoMSG_checksum(dst , pos);
    pos++;
    return pos;

}

void videoData_comAck(INT32 socket ,INT32 type , INT32 ack)
{
    UINT16 pos = 0;  
    UINT32 size = 0; 
    CHAR data[128] = {0};
    
    data[pos++] = _VIDEO_MSG_HEADER1_;
    data[pos++] = _VIDEO_MSG_HEADER2_;
    
    pos += 4;//skip len
    data[pos++] = type;
    pos++;  
    pos += fill_version(&data[pos]);      
    data[pos++] = ack;
    
    size = pos + 1;
    memcpy(&data[2] ,&size,sizeof(size)); 
    data[pos] = videoMSG_checksum(data , pos);
    pos++;
    videoMSG_send(socket , data ,pos); 
}
void videoMSG_comAck(INT32 socket ,INT32 type , INT32 ack)
{
    CHAR buf[65] = {0};
    INT32 len = 0;
    
    if((len = videoMSG_enCode(type ,ack ,NULL , 0 ,buf ,sizeof(buf))) > 0)
        videoMSG_send(socket , buf ,len);
    return;
}
INT32 queue_middle_write(VIDEOPACKET_INFO *dataQueue ,QUEUE_DESC *dataList ,INT32 channel)
{
    CHAR data[VIDEO_DATA_MAX_SIZE] = {0};
    INT32 size = 0;
    INT32 i = 0;

    if(dataQueue == NULL || dataList == NULL || channel <= 0){
        Error("<video> dataQueue == NULL || dataList == NULL || channel[%d] <= 0 ",channel);  
        return 0;
    }
    for(i = 0 ; i < dataQueue->size;i++){
        if(dataQueue->data[i].valid && dataQueue->data[i].len > 0){
            memcpy(&data[size],dataQueue->data[i].data,dataQueue->data[i].len);
            size += dataQueue->data[i].len;
        }
    }
    if(size > 0){
        queue_write(dataList, (void *)&channel ,data, size); 
        if(stMqttMsgInfo.consoleLog)    
            INFO("channel[%d] queue_write size %d",channel,size);                  
     }
     memset(dataQueue , 0 ,sizeof(VIDEOPACKET_INFO));  
     return 1;
}
INT32 videoData_insertQueue(VIDEOPACKET_INFO *dataQueue ,QUEUE_DESC *dataList,VIDEOPACKET_HEAD *header, CHAR *data ,UINT32 size,INT32 channel)
{
    INT32 valid = 0;
    INT32 qPos = 0;
    INT32 index = 0;
    INT32 mID = 0;
    pthread_t tid = pthread_self();
 
    if(dataQueue == NULL || header == NULL || data == NULL){
        Error("<video> videoData_insertQueue channel[%d]  dataQueue == NULL || header == NULL || data == NULL",channel);
        return 0;
    }
    if(header->index >= VIDEO_DATA_PACKET_BUFFER_COUNT || size > VIDEO_DATA_PACKET_BUFFER_SIZE){
        Error("<video> recv drop packet[%d,%d] ",index,size);
        return 0;
    }

    for(qPos = 0;qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){//inserted 
        if(dataQueue[qPos].id == header->id)       
            break;    
                           
    }   
    if(qPos >= VIDEO_DATA_PACKET_QUEUE_MAX_COUNT){
         for(qPos = 0;qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){//first insert
             if(!dataQueue[qPos].valid)
                 break;
         }
    }   
    if(qPos >= VIDEO_DATA_PACKET_QUEUE_MAX_COUNT){//full data
        for(qPos = 0;qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){//min id data
             if(mID <= 0){
                 mID = dataQueue[qPos].id;
             }else if(mID > dataQueue[qPos].id){
                mID = dataQueue[qPos].id;
             }
        }
        if(mID > 0){
             for(qPos = 0;qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){
                 if(dataQueue[qPos].id == mID){//min id
                    break;
                 }
             }
             if(qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT){//size < total
                  INFO("tid [%ld] big data id=%d,total=%d,size=%d,write file",tid,dataQueue[qPos].id,dataQueue[qPos].total,dataQueue[qPos].size); 
                  dataQueue[qPos].size = dataQueue[qPos].total;
                  queue_middle_write(&dataQueue[qPos] ,dataList, channel); 
             }             
        }
    }
   // qPos = header->id % VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;
    if(qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT){ 
        index = header->index;
        memcpy(dataQueue[qPos].data[index].data , data, size);   
        valid = dataQueue[qPos].data[index].valid;
        dataQueue[qPos].data[index].valid = 1;
        dataQueue[qPos].data[index].len = size;
        dataQueue[qPos].id = header->id;
        dataQueue[qPos].total = header->total;
        dataQueue[qPos].valid = 1;   
        if(!valid)
            dataQueue[qPos].size += 1;
    }
    return 1;
}
INT32 videoSubServerWrite_task(void *arg)
{
    VIDEO_DATA_PTHREAD_SHARE *shareInfo = NULL;
    INT32 channel = 0;   
    CHAR data[VIDEO_DATA_MAX_SIZE] = {0};    
   
    if(arg == NULL){
        Error ("<video> sub write pthread arg == NULL exit");  
        return -1;
    }
    shareInfo = (VIDEO_DATA_PTHREAD_SHARE *)arg;

    while(1){ 
        UINT32 len = 0;
        UINT32 size = 0;
        memset(data, 0, sizeof(data));    

        if(shareInfo->endFlg){
            if(channel > 0){
                videoFile_end(&videoDevInfo.videoCtrl.CHsctrl[channel - 1].file);
                Error ("<video> sub write pthread channel[%d] exit",channel);  
            }else{
                Error ("<video> sub write pthread unknown channel exit");  
            }
             
            shareInfo->endFlg = 0;   
            break;
        }
        if(channel > 0 && video15Min_check(channel) == OK){            
            videoFile_end(&videoDevInfo.videoCtrl.CHsctrl[channel - 1].file);                          
            videoFile_start(&videoDevInfo.videoCtrl.CHsctrl[channel - 1].file,channel,videoDevInfo.videoCtrl.CHsctrl[channel - 1].ch.name);                        
        }
        if(queue_curDataSize(&shareInfo->dataList) == 0){
            usleep(500000);//500ms
            continue;
        }
        while(1){
            len = queue_curDataSize(&shareInfo->dataList);
            if(len == 0)
                break;
            if((size + len) < sizeof(data))
                size += queue_read(&shareInfo->dataList , (void *)&channel ,&data[size]);//have memcpy
            else
                break;
        }
        
        if(channel <= 0){
            continue;
        }        
        if(videoDevInfo.videoCtrl.CHsctrl[channel - 1].file.fd <= STDERR_FILENO){
            videoFile_start(&videoDevInfo.videoCtrl.CHsctrl[channel - 1].file,channel,videoDevInfo.videoCtrl.CHsctrl[channel - 1].ch.name);
        }
        if(videoDevInfo.videoCtrl.CHsctrl[channel - 1].file.fd > STDERR_FILENO && size > 0){           
            write(videoDevInfo.videoCtrl.CHsctrl[channel - 1].file.fd , data,size);                    
        }    
    }
    return 0;
}
INT32 videoData_read(int socket , CHAR *data, int size )
{   
    UINT32 index = 0;
    UINT8 isMsg = 0;
    UINT32 msg_size = 0;
    UINT32 hSize = sizeof(video_msg_header) + 2;

    if(socket <= 0){
        return -1;
    }

    memset(data,0,size);  
    while (1)
    {       
	if(isMsg == 0){//帧头
	    if(index == hSize){		
                msg_size = *((UINT32*)&data[2]);                
                if(data[0] != (CHAR)_VIDEO_MSG_HEADER1_ || data[1] != (CHAR)_VIDEO_MSG_HEADER2_){
void_print(data,index,socket);
                    recv(socket, data,VIDEO_DATA_PACKET_MAX_SIZE,0);                  
                    return 1;
                }
                if(msg_size > size){
                    recv(socket, data,VIDEO_DATA_PACKET_MAX_SIZE,0);
                    return 1;
                }                
                isMsg = 1;  
	    }
	}else if(index == msg_size){            
            return msg_size;
        }
	INT32 len = 0;
        if(isMsg == 0 && index < hSize){//帧头
            len = recv(socket, &data[index], hSize - index,0);
            if(len <= 0){//<=
                if(errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
                    continue;			    
                Error("<video>Data recv header %d:%d socket error:%s(errno:%d)",index,len,strerror(errno),errno);
                return -1;
                // continue;
            }
            index += len;
         }else if(isMsg && (index < msg_size)){//数据
             len = recv(socket, &data[index], msg_size - index,0);
             if(len <= 0){	//<=	
                 if(errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN)
                    continue;		    
                 Error("<video>Data recv data %d:%d socket error:%s(errno:%d)",index,len,strerror(errno),errno);
                 return -1;
                 // continue;
             }
             index += len;
         }
    }
    return 1;
}
void videoDataServer_processData(VIDEODataChannel * dataChs,VIDEOPACKET_INFO *dataQueue ,VIDEO_DATA_PTHREAD_SHARE *shareInfo,CHAR * data,INT32 size)
{
    VIDEOPACKET_HEAD *header; 

    header = (VIDEOPACKET_HEAD *)&data[VIDEO_DATA_PACKET_HEADER_START_POS];//datalen start 8 byte
    if(stMqttMsgInfo.consoleLog)
        INFO("channel[%d] header: id=%d,%d,%d,%d",dataChs->channel,header->id,header->isPacket,header->total,header->index);
    if(!header->isPacket){//not packet
        if(dataChs->curPacketId == header->id){// no packet,curPackage
            queue_write(&shareInfo->dataList, (void *)&dataChs->channel ,&data[VIDEODATA_START_POS] , size - VIDEODATA_START_POS);
            dataChs->curPacketId++;
            if(stMqttMsgInfo.consoleLog)
                INFO("channel[%d] queue_write single size %d",dataChs->channel,size - VIDEODATA_START_POS); 
            return;
        }else if(dataChs->curPacketId < header->id){
            videoData_insertQueue(dataQueue,&shareInfo->dataList ,header,&data[VIDEODATA_START_POS] , size - VIDEODATA_START_POS,dataChs->channel);   
        }
    }else{//need insert queue,no packet or packet
        videoData_insertQueue(dataQueue ,&shareInfo->dataList,header,&data[VIDEODATA_START_POS] , size - VIDEODATA_START_POS,dataChs->channel);
    }
    while(1)
    {//check if write file  
        INT32 idCount = 0; 
        INT32 qPos = 0;
        INT32 i = 0;                      
        for(qPos = 0 ; qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){
            if(dataQueue[qPos].id == dataChs->curPacketId && dataQueue[qPos].valid){
                break;
            }
        }
        for(i = 0; i < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;i++){//data big id count
            if(dataQueue[i].id > 0 && dataQueue[i].valid)
            idCount++;                       
        }
        if(idCount == 0){
            break;
        }

        if(qPos >= VIDEO_DATA_PACKET_QUEUE_MAX_COUNT){
            INT32 mID = 0;
            for(qPos = 0;qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){//min id data
                if(mID <= 0){
                    mID = dataQueue[qPos].id;
                }else if(mID > dataQueue[qPos].id){
                    mID = dataQueue[qPos].id;
                }
            }
       
            for(qPos = 0;qPos < VIDEO_DATA_PACKET_QUEUE_MAX_COUNT;qPos++){
                if(dataQueue[qPos].id == mID){//min id
                    dataChs->curPacketId = mID;
                    break;
                }
            }                    
        }
        if(qPos >= VIDEO_DATA_PACKET_QUEUE_MAX_COUNT){//no data id
            break;
        }
        if(idCount > 2 && dataQueue[qPos].size < dataQueue[qPos].total && dataQueue[qPos].size > 0){//write file    
            if(dataQueue[qPos].id > 0){//drop packet
                dataQueue[qPos].size = dataQueue[qPos].total;
            }
        }
        if(dataQueue[qPos].id == dataChs->curPacketId && dataQueue[qPos].size >= dataQueue[qPos].total && dataQueue[qPos].size > 0){//write file  
            dataChs->curPacketId += queue_middle_write(&dataQueue[qPos] , &shareInfo->dataList, dataChs->channel);
        }else{
            break;
        }
    } 
    return;
}
INT32 videoDataServer_task (void *arg)
{
    VIDEODataChannel  dataChs[VIDEO_CHANNEL_MAX_COUNT];
    fd_set rfds;
    INT32 maxFd = -1;
    struct timeval timeout;
    INT32 i = 0;
    VIDEO_DATA_PTHREAD_SHARE shareInfo[VIDEO_CHANNEL_MAX_COUNT];
    void *packetInfo[VIDEO_CHANNEL_MAX_COUNT] = {0};
    
    memset(dataChs,0,sizeof(VIDEODataChannel) * VIDEO_CHANNEL_MAX_COUNT);
    if(stMqttMsgInfo.isUsbCollect){
        sleep(1);
        pthread_exit(NULL);
        return 0;
    }
    htkj_system("mkdir -p %s",VACVIDEO_BASEPATH);
    delEmptyDir(VACVIDEO_BASEPATH);
    delEmptyDir(VACVIDEO_BASEPATH);
    delEmptyDir(VACVIDEO_BASEPATH);
    while(1){
        if(videoDevInfo.sSocket <= 0 && videoDevInfo.dataTcp.port > 0){
            videoDevInfo.sSocket = TCPServer_init(videoDevInfo.dataTcp.port);            
            if(videoDevInfo.sSocket < 0){  
                if(i < 9){
                    i++;
                    sleep(1);
                }              
                Error("<video> Create Video Data Tcp Server[%d] error",videoDevInfo.dataTcp.port);                   
                return -1;
            }                    
            INFO("<video> Create Video Data Tcp Server[%d] ok",videoDevInfo.dataTcp.port);
            break;
        }
        usleep(10000);
    }
    while(1){
        INT32 ret = -1;
        timeout.tv_sec = 0;
        timeout.tv_usec = 20000;//20ms   
        
        for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT; i++){
            if(videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].ch.valid == VIDEOMSG_FAIL || videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].status != CMD_INIT){
                continue;
            }
            if(dataChs[i].pid > 0){
                shareInfo[i].endFlg = 1;
                pthread_join(dataChs[i].pid, NULL);                          
                INFO ("<video> Video Data subServer sub write pthread[%ld] socket[%d] exit !",dataChs[i].pid,dataChs[i].socket);           
                videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].status = CMD_INIT;
                queue_destroy(&shareInfo[i].dataList);
            }
            INFO ("<video> Video Data Server pthread socket[%d] timeout exit!",socket);
            close(dataChs[i].socket);            
            memset(&dataChs[i],0,sizeof(VIDEODataChannel));
            if(packetInfo[i]){
                free(packetInfo[i]);
                packetInfo[i] = NULL;
            }              
            continue;  
        }       

        FD_ZERO(&rfds); 
        maxFd = videoDevInfo.sSocket;
        FD_SET(videoDevInfo.sSocket,&rfds);
        for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT; i++){
            if(dataChs[i].socket > 0){
                FD_SET(dataChs[i].socket,&rfds);
                if(dataChs[i].socket > maxFd)
                    maxFd = dataChs[i].socket;
            }
        }//add read fd set
        ret = select(maxFd + 1, &rfds, NULL, NULL, &timeout);      
        if(ret < 0) {   
            Error("<video> Video Data Server[%d] select error:%s(errno:%d)",videoDevInfo.dataTcp.port,strerror(errno),errno) 
            continue;   
        }else if(ret == 0){            
            continue;
        }
        if(FD_ISSET(videoDevInfo.sSocket, &rfds)){
            socklen_t len = sizeof(struct sockaddr_in);
            struct sockaddr_in socket;
            INT32 client_socket;
            UINT8 flg = 0;

            client_socket = accept(videoDevInfo.sSocket,(struct sockaddr*)(&socket),&len);            
            if(client_socket < 0){
                Error("<video> Video Data Server[%d], accept client error:%s(errno:%d)",videoDevInfo.dataTcp.port,strerror(errno),errno);
                //if(WSAENETDOWN == errno){// || WSAECONNRESET == errno){
                    //down server
               // }
            }else{
                INT32 flags = 0;
                INT32 bufLen = VIDEO_DATA_SOCKET_SYS_CACHE_SIZE;
                INFO ("<video> Video Data Server[%d], socket[%d %s:%d] Ok!",videoDevInfo.dataTcp.port,client_socket,inet_ntoa(socket.sin_addr),ntohs(socket.sin_port));
                flags = fcntl(client_socket, F_GETFL, 0);                       //获取文件的flags值。
                fcntl(client_socket, F_SETFL, flags | O_NONBLOCK);
                setsockopt(client_socket,SOL_SOCKET,SO_RCVBUF,&bufLen,sizeof(bufLen));
                for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT ; i++ ){
                    if(dataChs[i].socket <= 0){
                        flg = 1;
                        dataChs[i].socket = client_socket;
                        break;
                    }
                }
                if(flg == 0){
                    Error("<video> Video Data Server[%d], max[%d] client full!",videoDevInfo.dataTcp.port,VIDEO_CHANNEL_MAX_COUNT);
                    close(client_socket);
                }
            }
        }
        for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT ; i++){
            CHAR recv_buf[VIDEO_DATA_MAX_SIZE] = {0}; 
            INT32 len = 0;
            if(dataChs[i].socket <= 0)
                continue;
            if(!FD_ISSET(dataChs[i].socket, &rfds))
                continue;
            //len = videoData_read(dataChs[i].socket , recv_buf, sizeof(recv_buf) );
            len = videoData_recv(dataChs[i].socket , recv_buf, sizeof(recv_buf) );
            if(len < 0){
                if(dataChs[i].pid > 0){
                    shareInfo[i].endFlg = 1;
                    pthread_join(dataChs[i].pid, NULL);                          
                    INFO ("<video> Video Data subServer sub write pthread[%ld] socket[%d] exit !",dataChs[i].pid,dataChs[i].socket);           
                    videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].status = CMD_INIT;
                    queue_destroy(&shareInfo[i].dataList);
                }
                INFO ("<video> Video Data Server pthread socket[%d] exit!",socket);
                close(dataChs[i].socket);
                FD_CLR(dataChs[i].socket, &rfds);
                memset(&dataChs[i],0,sizeof(VIDEODataChannel));
                if(packetInfo[i]){
                    free(packetInfo[i]);
                    packetInfo[i] = NULL;
                }              
                continue;   
            }else if(len > VIDEO_DATA_CHANNEL_POS){              
                if(dataChs[i].channel <= 0 && recv_buf[VIDEO_DATA_TYPE_POS] == MSG_VIDEO_DATA_CHANNEL_NEW){//0x12
                    if(recv_buf[VIDEO_DATA_CHANNEL_POS] > 0 && recv_buf[VIDEO_DATA_CHANNEL_POS] <= VIDEO_CHANNEL_MAX_COUNT){
                        dataChs[i].channel = recv_buf[VIDEO_DATA_CHANNEL_POS];
                    }else{
                        videoData_comAck(dataChs[i].socket,MSG_VIDEO_DATA_CHANNEL_NEW_ACK , VIDEOMSG_FAIL);
                        continue;
                    }    
                    //start write file pthread
                    //malloc cache queue       
                    if(OK != queue_init (&shareInfo[i].dataList , sizeof(INT32),VIDEO_DATA_MAX_COUNT , VIDEO_DATA_MAX_SIZE)){
                        Error("<video> Video Data Server channel[%d] queue_init error",dataChs[i].channel);
                        videoData_comAck(dataChs[i].socket,MSG_VIDEO_DATA_CHANNEL_NEW_ACK , VIDEOMSG_FAIL);
                        videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].status = CMD_INIT;   
                        close(dataChs[i].socket);
                        FD_CLR(dataChs[i].socket, &rfds);
                        memset(&dataChs[i],0,sizeof(VIDEODataChannel));
                        continue;
                    }
                    packetInfo[i] = malloc(sizeof(VIDEOPACKET_INFO) * VIDEO_DATA_PACKET_QUEUE_MAX_COUNT);
                    if(packetInfo[i] == NULL){
                        Error ("<video> Video Data Server channel[%d] malloc packet cache error:%d(%s)",dataChs[i].channel,errno,strerror(errno));  
                        videoData_comAck(dataChs[i].socket,MSG_VIDEO_DATA_CHANNEL_NEW_ACK , VIDEOMSG_FAIL);
                        videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].status = CMD_INIT;
                        close(dataChs[i].socket);
                        FD_CLR(dataChs[i].socket, &rfds);
                        memset(&dataChs[i],0,sizeof(VIDEODataChannel));                        
                        queue_destroy(&shareInfo[i].dataList);
                        continue;
                    }
                    memset(packetInfo[i], 0 ,sizeof(VIDEOPACKET_INFO) * VIDEO_DATA_PACKET_QUEUE_MAX_COUNT);
                    if (0 != pthread_create (&dataChs[i].pid, NULL , (void *)videoSubServerWrite_task, &shareInfo[i])){
                        Error ("<video> Video Data Server channel[%d] new write pthread error:%d(%s)",dataChs[i].channel,errno,strerror(errno));
                        videoData_comAck(dataChs[i].socket,MSG_VIDEO_DATA_CHANNEL_NEW_ACK , VIDEOMSG_FAIL);
                        videoDevInfo.videoCtrl.CHsctrl[dataChs[i].channel - 1].status = CMD_INIT;
                        close(dataChs[i].socket);
                        FD_CLR(dataChs[i].socket, &rfds);
                        memset(&dataChs[i],0,sizeof(VIDEODataChannel));
                        if(packetInfo[i]){
                            free(packetInfo[i]);
                            packetInfo[i] = NULL;
                        }
                        queue_destroy(&shareInfo[i].dataList);
                        continue;
                    } 
                    INFO ("<video> Video Data Server channel[%d] new write pthread OK!",dataChs[i].channel);
                     
                    videoData_comAck(dataChs[i].socket,MSG_VIDEO_DATA_CHANNEL_NEW_ACK , VIDEOMSG_OK);
                    continue;
                }else if(recv_buf[VIDEO_DATA_TYPE_POS] == MSG_VIDEO_DATA && dataChs[i].channel > 0){//x014 video data
                    videoDataServer_processData(&dataChs[i],(VIDEOPACKET_INFO *)packetInfo[i] ,&shareInfo[i],recv_buf , len);
                }
            }
        } 
    }   
    close(videoDevInfo.sSocket);      
    videoDevInfo.sSocket = -1;  

    return 0;
}

int videoMSG_check(video_msg_header *header)
{
    if(header->head1 != (CHAR)_VIDEO_MSG_HEADER1_){
        return ERROR;
    }
    if(header->head2 != (CHAR)_VIDEO_MSG_HEADER2_){
        return ERROR;
    }

    return OK;
}
void magClient_close(void)
{
    INT32 i = 0;
    if(videoDevInfo.socket > 0)
        close(videoDevInfo.socket);
    videoDevInfo.socket = -1;
    for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT; i++ ){
        videoDevInfo.videoCtrl.CHsctrl[i].status = CMD_INIT;                
    }
}

INT32 videoData_recv(int socket , CHAR *data, int size )
{
    struct timeval timeout={0,200}; //1s timeout
    fd_set rfds;
    UINT32 index = 0;
    UINT8 isMsg = 0;
    UINT32 msg_size = 0;
    UINT32 cTime = currentTimeGet_int();
    UINT32 hSize = sizeof(video_msg_header) + 2;

    if(socket <= 0){
        return -1;
    }

    memset(data,0,size);    
	
    while (1)
    {	
        UINT32 curTime = currentTimeGet_int();

        if((curTime - cTime) > 1){//VIDEODEV_timeout
            return -1;
        }
        timeout.tv_sec = 0;
        timeout.tv_usec = 200000;//200ms
	if(isMsg == 0){//帧头
	    if(index == hSize){		
                msg_size = *((UINT32*)&data[2]);                
                if(data[0] != (CHAR)_VIDEO_MSG_HEADER1_ || data[1] != (CHAR)_VIDEO_MSG_HEADER2_){
void_print(data,index,socket);
                    //memcpy(data,&data[1] , hSize - 1);
                    index = 0;
                    continue;
                }
                if(msg_size > size){
                    //memcpy(data,&data[1] , hSize - 1);
                    index = 0;
                    continue;
                }                
                isMsg = 1;  
	    }
	}else if(index == msg_size){            
            return msg_size;
        }
	FD_ZERO(&rfds); //每次循环都要清空集合，否则不能检测描述符变化  
        FD_SET(socket,&rfds); //添加描述符
	
	switch(select(socket+1,&rfds,NULL,NULL,&timeout))   //select使用  
        {  
            case -1: 
                Error("<video>Data select -1 error:%s(errno:%d)",strerror(errno),errno);
                return -1;
                //break; //select错误，退出程序  
            case 0:               
                break; //再次轮询  
            default:  
                if(FD_ISSET(socket,&rfds)) //测试sock是否可读，即是否网络上有数据  
                { 
                    INT32 len = 0;
                    if(isMsg == 0 && index < hSize){//帧头
                        len = recv(socket, &data[index], hSize - index,0);
                        if(len <= 0){//<=			    
                            Error("<video>Data recv header %d:%d socket error:%s(errno:%d)",index,len,strerror(errno),errno);
                            return -1;
                           // continue;
                        }
                        index += len;
                    }else if(isMsg && (index < msg_size)){//数据
                        len = recv(socket, &data[index], msg_size - index,0);
                        if(len <= 0){	//<=		    
                            Error("<video>Data recv data %d:%d socket error:%s(errno:%d)",index,len,strerror(errno),errno);
                            return -1;
                           // continue;
                        }
                         index += len;
                    }
                }      
                break;
         }
    }
    return 0;
}
//socket, hSize = headerSize,lPos = pos of len in frame ,lSize =size of type of len ,data = recv buffer,size = sizeof(data)

INT32 videoMSG_recv(int socket ,CHAR *data, int size )
{
    struct timeval timeout={0,200}; //1s timeout
    fd_set rfds;
    UINT32 index = 0;
    UINT8 isMsg = 0;
    UINT32 msg_size = 0;
    UINT32 cTime = currentTimeGet_int();
    UINT32 hSize = sizeof(video_msg_header);

    if(socket <= 0){
        return -1;
    }   
    memset(data,0,size);    
	
    while (1)
    {	
        UINT32 curTime = currentTimeGet_int();

        if((curTime - cTime) > 1){//VIDEODEV_timeout
     //       return 0;
        }
        timeout.tv_sec = 0;
        timeout.tv_usec = 200000;//200ms
	if(isMsg == 0){//帧头
	    if(index == hSize){		
                msg_size = *((UINT16*)&data[2]);                
                if(data[0] != (CHAR)_VIDEO_MSG_HEADER1_ || data[1] != (CHAR)_VIDEO_MSG_HEADER2_){
void_print(data,index,socket);
                    memcpy(data,&data[1] , hSize - 1);
                    index--;
                    continue;
                }
                if(msg_size > size){
                    memcpy(data,&data[1] , hSize - 1);
                    index--;
                    continue;
                }                
                isMsg = 1;  
	    }
	}else if(index == msg_size){
            INFO("<video>MSG<== type[%d],size[%d],SC[%x],%s",data[4],msg_size,data[msg_size-1],&data[6]);            
            void_print(data,index,socket);
            return msg_size;
        }
	FD_ZERO(&rfds); //每次循环都要清空集合，否则不能检测描述符变化  
        FD_SET(socket,&rfds); //添加描述符
	
	switch(select(socket+1,&rfds,NULL,NULL,&timeout))   //select使用  
        {  
            case -1: 
                Error("<video>MSG select -1 error:%s(errno:%d)",strerror(errno),errno);
                return -1;
                //break; //select错误，退出程序  
            case 0:               
                break; //再次轮询  
            default:  
                if(FD_ISSET(socket,&rfds)) //测试sock是否可读，即是否网络上有数据  
                { 
                    INT32 len = 0;
                    if(isMsg == 0 && index < hSize){//帧头
                        len = recv(socket, &data[index], hSize - index,0);
                        if(len <= 0){//<=			    
                            Error("<video>MSG recv header %d:%d socket error:%s(errno:%d)",index,len,strerror(errno),errno);
                            return -1;
                           // continue;
                        }
                        index += len;
                    }else if(isMsg && (index < msg_size)){//数据
                        len = recv(socket, &data[index], msg_size - index,0);
                        if(len <= 0){	//<=		    
                            Error("<video>MSG recv data %d:%d socket error:%s(errno:%d)",index,len,strerror(errno),errno);
                            return -1;
                           // continue;
                        }
                         index += len;
                    }
                }      
                break;
         }
    }
    return 0;
}
void videoDev_readConfig(void)
{
    CHAR filename[LINE_CONTENT_MAX_LEN] = {0};
    CHAR tmp[LINE_CONTENT_MAX_LEN] = {0};
    INT32 i = 0;
    INT32 size = 0;

    com_getKeyValue("sysConfFile",filename,stMqttMsgInfo.appConfig);
    if(strlen(filename) < 5){
        Error("<video> sysConfFile [%s] not empty!\n",filename);
        strcpy(filename,SYS_DEFAULT_CONFIG);
    }
    if(0 == access(filename, F_OK )){
        if(0 != access(filename, R_OK | W_OK)){
            INFO("<video> %s need read-write permission!\n",filename);
            htkj_system("chmod 777 %s",filename);
        }
    }
    videoDevInfo.msgTcp.port = atoi(com_getKeyValue("videoMsg_server_port",tmp,filename));   
    com_getKeyValue("videoMsg_server_ip",videoDevInfo.msgTcp.ip ,filename);
    videoDevInfo.dataTcp.port = atoi(com_getKeyValue("videoData_server_port1",tmp,filename));
     
    if(videoDevInfo.msgTcp.port <= 0)
        videoDevInfo.msgTcp.port = VIDEO_MSG_SERVER_PORT;
    if(videoDevInfo.dataTcp.port <= 0)
        videoDevInfo.dataTcp.port = VIDEO_DATA_PORT;
    if(strlen(videoDevInfo.msgTcp.ip) < 7)
        strcpy(videoDevInfo.msgTcp.ip ,VIDEO_MSG_SERVER_IP);    
        
    get_localIPByRemoteIP (videoDevInfo.msgTcp.ip,videoDevInfo.dataTcp.ip,24);
    
    if(strlen(videoDevInfo.dataTcp.ip) != 15){//不足15位后补空格
        size = 15 - strlen(videoDevInfo.dataTcp.ip);
        for(i = 0 ; i < size ; i++)
        {
            videoDevInfo.dataTcp.ip[strlen(videoDevInfo.dataTcp.ip)] = ' ';
        }
    }
    return;
}
INT32 videoMSG_deCode(const CHAR*data , INT32 size ,CHAR *dst , INT32 *type )
{
    INT32 pos = 0;
    INT32 len = 0;
    video_msg_header *header = NULL;   
    CHAR ver[10] = {0}; 
    CHAR checksum = 0;
    
    if(data == NULL || size < sizeof(video_msg_header)){
        return ERROR;
    }
    header = (video_msg_header*)data;

    if(videoMSG_check(header) != OK){
        Error("<video> msg check header error[0x%x,0x%x]",header->head1,header->head2);
        return ERROR;
    }
    checksum = videoMSG_checksum(data,size - 1);
    if(header->size != size){
        Error("<video> msg check size error[0x%x!=0x%x]",size,header->size);
        return ERROR;
    }
    if(data[size - 1] != checksum){
        Error("<video> msg check checksum error[0x%x!=0x%x]",checksum,data[size - 1]);
        return ERROR;
    }
    pos += sizeof(video_msg_header);   
    *type = data[pos++];
    pos++;
    memcpy(ver,&data[pos],VIDEO_VERSION_LENGHT);
    pos += VIDEO_VERSION_LENGHT;
    len = size - pos - 1;
    if(len <= 0){
        return ERROR;
    }   
    memcpy(dst,&data[pos] ,len);
    return len;  
}
void trimString(CHAR *data ,CHAR *src)
{
    if(data == NULL || src == NULL)
        return;
     
    //memcpy(buf,data,len);
   
    while(*src != '\0'){
        if(*src != ' '){
            *data = *src;
            data++;
        }
        src++;
    }
    return;  
}
INT32 decode_mfrs(CHAR *data, INT32 len)
{
    INT32 i = 0;
    CHAR buf[32] = {0};

    if(data == NULL || len <= 0)
       return ERROR;
    
    videoDevInfo.mfrs.ID = data[0];
    if(videoDevInfo.mfrs.ID <= 0)
        return ERROR;
    len--;
    if(len <= 0)
       return ERROR;
    while (1){
        if(MFRS_lists[i].ID == (CHAR)LAST_NULL ){
            strcpy(videoDevInfo.mfrs.name , MFRS_lists[i].name); 
            return ERROR;
        }
        if(MFRS_lists[i].ID == videoDevInfo.mfrs.ID){
           strcpy(videoDevInfo.mfrs.name , MFRS_lists[i].name); 
           break;
        }
        i++;
    }       
   
    memcpy(buf ,&data[1] , strlen(&data[1]));
    trimString(videoDevInfo.train.ID ,buf);
   // strcpy(videoDevInfo.train.ID,buf);
    sprintf(videoDevInfo.backPath,"%s/6A-VIDEO-%s-%s",stMqttMsgInfo.uploadLinkPath,videoDevInfo.train.ID,videoDevInfo.mfrs.name);//store file
    sprintf(videoDevInfo.path,"%s/6A-VIDEO-%s-%s",VACVIDEO_BASEPATH,videoDevInfo.train.ID,videoDevInfo.mfrs.name);//tmp store file
    INFO("<video> decode mfrs path:%s",videoDevInfo.path);
    return OK;
}
INT32 decode_validChannel(UINT16 chs)
{
    INT32 i = 0;
    INT32 flg = ERROR;
    INFO("<video> channel chs[%d]",chs);
    for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT; i++){
        if((chs >> i) & 0x01){
            videoDevInfo.videoCtrl.CHsctrl[i].ch.valid = VIDEOMSG_OK;
            videoDevInfo.videoCtrl.validSize++;
            flg = OK;
            INFO("<video> channel id[%d]",i+1);
        }
    }
    return flg;
}
INT32 videoMSG_process(INT32 socket , CHAR *data, INT32 size)
{
    INT32 type = 0;
    CHAR dst[VIDEO_MSG_MAX_SIZE] = {0};
    INT32 len = 0;   
    INT32 id = 0;
 
    if(socket < 0)
        return ERROR;
    len = videoMSG_deCode(data , size ,dst , &type );
    if(len == ERROR){
        if(data[4] == MSG_MFRS_ID && size == 64){//com data msg
            //queue_write (&videoMsgDesc ,NULL, data, 64);
            return ERROR;
        }
        Error("<video> MSG decode error type[%d]",data[4]);
        return ERROR;
    }
    switch(type){
        case MSG_REG_ACK:
             if(dst[0] == VIDEOMSG_OK)//reg ok
                 videoDevInfo.regStatus = CMD_OK;
             else
                 videoDevInfo.regStatus = CMD_INIT;
        break;
        case MSG_MFRS_ID:
             if(decode_mfrs(dst , len) == OK){
                 videoDevInfo.idStatus = CMD_OK;       
              }
              videoMSG_comAck(socket,MSG_MFRS_ID_ACK , VIDEOMSG_OK);
        break;
        case MSG_STATUS_ACK:
           //  if(dst[0] == VIDEOMSG_OK)//heart ok
                 videoDevInfo.hStatus = CMD_OK;
            // else
              //   videoDevInfo.hStatus = CMD_INIT;
        break;
        case MSG_GET_VIDEO_CHANNEL_ACK:
             if(dst[0] == VIDEOMSG_OK){//valid channel info 
                 if(decode_validChannel(*((UINT16*)&dst[1])) == OK)
                     videoDevInfo.videoCtrl.status = CMD_OK;
             }else
                 videoDevInfo.videoCtrl.status = CMD_INIT;    
        break;
        case MSG_REQ_VIDEO_CHANNEL_ACK:
             id = dst[1];
             if( id > 0)
                 id--;
             if(dst[0] == VIDEOMSG_OK){//request video channel act  
                 videoDevInfo.videoCtrl.CHsctrl[id].status = CMD_OK;  
             }else
                 videoDevInfo.videoCtrl.CHsctrl[id].status = CMD_INIT;
        break;
        case MSG_COM_SAFE:
             queue_write (&videoMsgDesc ,NULL, dst, len);
        break;
        default:
        break;		
    }
    return OK;
}
INT32 videoDev_init(void)
{
    int i = 0;
    pthread_t pid;
    
    memset(&videoDevInfo,0,sizeof(videoDevInfo));
    videoDev_readConfig();  
    queue_init (&videoMsgDesc, 0 ,VIDEO_MSG_IDEX_BUFFER_COUNT , VIDEO_MSG_IDEX_BUFFER_MAX_SIZE) ;
    
    while (1){
        if(CHALLEL_lists[i].ID == (CHAR)LAST_NULL )
            break;
        videoDevInfo.videoCtrl.CHsctrl[i].ch.ID = CHALLEL_lists[i].ID;
        strcpy(videoDevInfo.videoCtrl.CHsctrl[i].ch.name , CHALLEL_lists[i].name);
        i++;
    }    
    if (0 != pthread_create (&pid, NULL , (void *)idexProcess_task, NULL)){
         Error ("<video> Video Client  record com idx file, Create pthread Fail! error:%s(errno:%d)!",strerror(errno),errno);
         return ERROR;
               
     } 
     INFO ("<video> Video Client  record com idx file, Create pthread OK!");
     return OK;
}
INT32 videoMSG_send(INT32 socket , CHAR *data, INT32 size)
{
    INT32 len = 0;
    INT32 pos = 0;

    if(socket <= 0){
        Error("<video>MSG send data socket[%d] <= 0 ,error" ,socket );
        return ERROR;
    }   
    if(data == NULL || size < VIDEO_MSG_MIN_SIZE){
        Error("<video>MSG %d send data size[%d] < [%d]MSG min size" ,socket,size , VIDEO_MSG_MIN_SIZE);
        return 0;
    }
    while(pos < size){
	len = send(socket,&data[pos],size - pos,0);
	if(len <= 0){
            if ((errno == EINTR) || (errno == EAGAIN)){
                usleep(10*1000);/**发送区为满时候循环等待，直到可写为止*/
                continue;
            }
            Error("<video>MSG %dsend data pos[%d] error:%s(errno:%d)", socket,len,strerror(errno),errno);
            return ERROR;
	}
	pos += len;
    }
    
    data[size-1] = 0;
    if(socket == videoDevInfo.socket){
        INFO("<video>MSG %d==> type[%d],size[%d],SC[%x],%s" ,socket,data[4],size,data[size-1],&data[6]);
    }else{
        INFO("<video>Data %d==> type[%d],size[%d],SC[%x],%s" ,socket,data[6],size,data[size-1],&data[8]);
    }
    void_print(data,size,0);
    return size;
}
INT32 videoMSG_heartStatus(INT32 socket)
{
    CHAR buf[VIDEO_MSG_MAX_SIZE] = {0};
    CHAR tBuf[100] = {0};
    INT32 len = 0;
    
    tBuf[len++] = VIDEO_DEVICE_SELF_MFRS_ID;
    tBuf[len++] = VIDEOMSG_OK;
    tBuf[len++] = VIDEOMSG_OK;//--tmp--SSDstatus
    if(stMqttMsgInfo.link == IOT_STATUS_CONNECT)//stMqttMsgInfo.regHostAck == MQTT_REG_HOST_MSG_OK
        tBuf[len++] = VIDEOMSG_OK;
    else
        tBuf[len++] = VIDEOMSG_FAIL;
    tBuf[len++] = stMqttMsgInfo.dataSync;
    tBuf[len++] = stMqttMsgInfo.percent;//--tmp--
    if(isGroup() < 0)
      tBuf[len++] = VIDEOMSG_OK;
    else
      tBuf[len++] = VIDEOMSG_FAIL;

    if((len = videoMSG_enCode(MSG_STATUS ,-1 ,tBuf , len ,buf ,sizeof(buf))) > 0){
        return videoMSG_send(socket , buf ,len);
    }
    return 0;
}

INT32 videoMSG_reqChannel(INT32 socket , INT32 id)
{
    CHAR buf[VIDEO_MSG_MAX_SIZE] = {0};
    CHAR tBuf[100] = {0};
    INT32 len = 0;
  
    tBuf[len++] = videoDevInfo.videoCtrl.CHsctrl[id].ch.ID;        
    memcpy(&tBuf[len],videoDevInfo.dataTcp.ip,strlen(videoDevInfo.dataTcp.ip));
    len += strlen(videoDevInfo.dataTcp.ip);
    sprintf(&tBuf[len],"%04d",videoDevInfo.dataTcp.port);
    len += 4; 
      
    if((len = videoMSG_enCode(MSG_REQ_VIDEO_CHANNEL ,-1 ,tBuf , len ,buf ,sizeof(buf))) > 0)
        return videoMSG_send(socket , buf ,len);   
    return 0;
}
INT32 videoMSG_getChannel(INT32 socket)
{
    CHAR buf[VIDEO_MSG_MAX_SIZE] = {0};
    INT32 len = 0;
 
    if((len = videoMSG_enCode(MSG_GET_VIDEO_CHANNEL ,-1 ,NULL , 0 ,buf ,sizeof(buf))) > 0)
        return videoMSG_send(socket , buf ,len);
    return 0;
}
INT32 videoMSG_reg(INT32 socket)
{
    CHAR buf[VIDEO_MSG_MAX_SIZE] = {0};
    INT32 len = 0;
 
    if((len = videoMSG_enCode(MSG_REG ,-1 ,VIDEO_DEVICE_ID , strlen(VIDEO_DEVICE_ID) ,buf ,sizeof(buf))) > 0)
        return videoMSG_send(socket , buf ,len);
    return 0;
}
void oldFile_check(void)
{
    if(access("/tmp/VACinit_check",F_OK) != 0){
        htkj_system("touch /tmp/VACinit_check");
        dirBySuffix_link(VACVIDEO_BASEPATH,stMqttMsgInfo.uploadLinkPath,".tmp");
    }//VCAVIDEO_BASEPATH  /upload/VCAVideo
}
INT32 videoMSGClient_task (void *arg)
{
   // UINT32 cTime  = 0;
    UINT32 comTime = 0;
    UINT32 hTime  = 0;
    UINT32 curTime = 0;
    UINT32 hTimeout_base = 0;    
    CHAR recv_buf[VIDEO_MSG_MAX_SIZE] = {0};
    INT32  i = 0;
    if(stMqttMsgInfo.isUsbCollect){
        sleep(1);
        pthread_exit(NULL);
        return 0;
    }
    oldFile_check();
    videoDev_init();

    while(1){
        INT32 len = 0;

        usleep(50000);//50ms
        curTime = currentTimeGet_int();  
        if(videoDevInfo.socket <= 0){//(curTime - cTime) > VIDEO_REGISTER_timeout &&
            INT32 bufLen = VIDEO_MSG_SOCKET_SYS_CACHE_SIZE;
            videoDevInfo.socket = create_client(videoDevInfo.msgTcp.ip , videoDevInfo.msgTcp.port);           
            if(videoDevInfo.socket < 0){ 
               // if((curTime - cTime) > 300){ //~5min
                    //Error("<video> connect ip[%s:%d] error",videoDevInfo.msgTcp.ip, videoDevInfo.msgTcp.port);
                   // cTime = curTime;
               // } 
                continue;  
            }          
            INFO("<video> connect ip[%s:%d] ok",videoDevInfo.msgTcp.ip, videoDevInfo.msgTcp.port);
            setsockopt(videoDevInfo.socket,SOL_SOCKET,SO_RCVBUF,&bufLen,sizeof(bufLen));
            //cTime = curTime;
            comTime = 0;
        }
        
        if(((curTime - comTime) > VIDEODEV_timeout) && (comTime > 0)){ //timeout check
            if(videoDevInfo.regStatus != CMD_OK){//register timeout,repeat register 
                Error("<video> send register not ack timeout");
                magClient_close();
                videoDevInfo.regStatus = CMD_INIT;
                videoDevInfo.hStatus = CMD_INIT;               
                continue;             
            }
            if(videoDevInfo.videoCtrl.status == CMD_WAITTING){//get channel timeout
                videoDevInfo.videoCtrl.status = CMD_INIT;
            }else { //request channel timeout
                for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT; i++ ){
                    if(videoDevInfo.videoCtrl.CHsctrl[i].ch.valid == VIDEOMSG_FAIL || videoDevInfo.videoCtrl.CHsctrl[i].status == CMD_OK)
                        continue;
                    if(videoDevInfo.videoCtrl.CHsctrl[i].status == CMD_WAITTING)
                        videoDevInfo.videoCtrl.CHsctrl[i].status = CMD_INIT;
                
                }
            }
        }    
        if(videoDevInfo.regStatus == CMD_INIT){//register
            if(videoMSG_reg(videoDevInfo.socket) < 0){//send reg msg
                Error("<video> send register MSG socket error");
                magClient_close(); 
                continue;
            }
            videoDevInfo.regStatus = CMD_WAITTING;
            comTime = curTime;
        } 
        if(videoDevInfo.regStatus == CMD_OK && videoDevInfo.videoCtrl.status == CMD_INIT){//get channel
            if(strlen(videoDevInfo.train.ID) > 0){
                if(videoMSG_getChannel(videoDevInfo.socket) < 0){//send get channel msg
                    Error("<video> send get channel MSG socket error");
                    magClient_close(); 
                    continue;
                }
                videoDevInfo.videoCtrl.status = CMD_WAITTING;
                comTime = curTime;
            }
        }
        if(videoDevInfo.regStatus == CMD_OK && videoDevInfo.videoCtrl.status == CMD_OK && videoDevInfo.videoCtrl.validSize > 0 ){//request channel 
            for(i = 0 ; i < VIDEO_CHANNEL_MAX_COUNT; i++ ){
                if(videoDevInfo.videoCtrl.CHsctrl[i].ch.valid == VIDEOMSG_FAIL || videoDevInfo.videoCtrl.CHsctrl[i].status != CMD_INIT)
                    continue;
                if(videoMSG_reqChannel(videoDevInfo.socket,i) < 0){//send request channel msg
                    Error("<video> send request channel MSG socket error");
                    magClient_close(); 
                    continue;
                }
                videoDevInfo.videoCtrl.CHsctrl[i].status = CMD_WAITTING;
                comTime = curTime;
            }      
        }    
        if(videoDevInfo.regStatus == CMD_OK && (videoDevInfo.hStatus == CMD_INIT || videoDevInfo.hStatus == CMD_OK))    
            hTimeout_base = curTime;
        if((videoDevInfo.regStatus == CMD_OK) 
            && (
                 ((curTime - hTime) > VIDEO_HEARTBEAT_interval) //30 sec
                 || (stMqttMsgInfo.dataSync && (curTime - hTime) > 0)//uploading, and 1 sec
               )
          ){//when uploading,1 per 1sec
            if((curTime - hTimeout_base) > VIDEO_HEARTBEAT_timeout && (videoDevInfo.hStatus == CMD_WAITTING)){//heart timeout
                Error("<video> heart status timeout, need repead register");
                magClient_close(); 
                videoDevInfo.regStatus = CMD_INIT; //need repeat register
                videoDevInfo.hStatus = CMD_INIT;
                hTime = 0;                
                continue;
            }
            videoDevInfo.hStatus = CMD_WAITTING;             
            hTime = curTime; 
            if(videoMSG_heartStatus(videoDevInfo.socket) < 0){//send heart msg ethor CMD_WAITTING or CMD_OK
                Error("<video> send heart status MSG socket error");
                magClient_close(); 
                continue;
            }                 
        }
    
        len = videoMSG_recv(videoDevInfo.socket ,recv_buf, sizeof(recv_buf) );
        if(len < 0){//repeat connect not register
            Error("<video> recv MSG socket error");
            magClient_close(); 
            continue;
        }else if(len > 0){
             videoMSG_process(videoDevInfo.socket , recv_buf ,len );           
        }         
     }   
    return 0;
}
void systemTime_check(const CHAR *data ,INT32 size)
{    
    UINT32 dTime = 0;
    UINT8 YYYY = 0;
    UINT8 MM = 0;
    UINT8 DD = 0;
    UINT8 hh = 0;
    UINT8 mm = 0;
    UINT8 ss = 0;
    UINT32 ct = currentTimeGet_int();
    CHAR st[20] = {0};
    static CHAR once = 0;

    if(data == NULL || size < 11){
        return;
    }
    
    dTime = *((UINT32*)&data[5]);

    YYYY = (dTime >> 26) & 0x3F;
    MM = (dTime >> 22) & 0x0F;
    DD = (dTime >> 17) & 0x1F;
    hh = (dTime >> 12) & 0x1F;
    mm = (dTime >> 6) & 0x3F;
    ss = dTime & 0x3F;
    /*
    YYYY = data[5];
    MM = data[6];
    DD = data[7];
    hh = data[8];
    mm = data[9];
    ss = data[10];
    */
    sprintf(st,"%04d-%02d-%02d %02d:%02d:%02d",YYYY + 2000,MM,DD,hh,mm,ss);
    
    if((!once && ct != strTime2Int(st)) || ct < strTime2Int(st)){
        if(YYYY <= 0 || MM <= 0 || MM > 12 || DD <= 0 || DD > 31)
            return;
        if(hh < 0 || hh > 23 || mm < 0 || mm > 59 || ss < 0 || ss > 59)
            return;
        INFO("common cmd time %s",st);
        htkj_system("date -s '%s' ; hwclock -w ",st);   
        once = 1;     
    }

}

INT32 idexProcess_task (void *arg)
{
    UINT32 tTime = 0;
    CHAR buf[VIDEO_MSG_MAX_SIZE] = {0};
    UINT32 curTime = 0;
    INT32 len = 0;

    while(1){
        usleep (200000);//500ms
        curTime = currentTimeGet_int();
        memset (buf, 0x00, sizeof(buf));
        if((len = queue_read(&videoMsgDesc ,NULL ,buf)) <= 0){
            continue;
        }
        
        if((curTime - tTime)  > 1800){
            systemTime_check(buf ,len);            
            tTime = curTime;
        }
        if(videoDevInfo.videoCtrl.idexFile.fd <= STDERR_FILENO){ 
            videoIdxFile_start(&videoDevInfo.videoCtrl.idexFile);
        }
	if(videoDevInfo.videoCtrl.idexFile.fd > STDERR_FILENO){
            write(videoDevInfo.videoCtrl.idexFile.fd ,buf ,len);           
        } 
        if(videoNextDay_check() == OK){           
            videoFile_end(&videoDevInfo.videoCtrl.idexFile);
        }          
    }
    return OK;
}
